package com.giw.project.simple_graph_traversal;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TraversalMapper extends Mapper<LongWritable, Text, Text, Text>{

	private final static String tmp_path="tmp.txt";

	protected void map(LongWritable key, Text value, Context context)
	throws IOException, InterruptedException{

		List<String> currentNodes=new LinkedList<String>();
		File f=new File(tmp_path);
		FileInputStream fis=new FileInputStream(f);
		InputStreamReader is=new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line=buffer.readLine();
		boolean isLitteralNode=!(line.split("\t")[0].contains("http"));
		while(line!=null){
			line=line.split("\t")[0];
			if(isLitteralNode) line = line.toLowerCase();
			currentNodes.add(line);
			line=buffer.readLine();
		}
		buffer.close();
		is.close();
		fis.close();
		Collections.sort(currentNodes);
		line=value.toString();
		String node=line.split("\t")[0];
		if(isLitteralNode) node = node.toLowerCase();
		String actualNode="";
		if(node.length()>=	2){
			actualNode = node.substring(1,node.length()-1);
			if(isLitteralNode) actualNode=actualNode.toLowerCase();
		}
		if(Collections.binarySearch(currentNodes,node)>=0||
				Collections.binarySearch(currentNodes,actualNode)>=0){
			String a = line.split("\t")[1];
			context.write(new Text(node),new Text(a));
		}

	}
}
